package com.mate.cloud;

import com.mate.cloud.listener.TransactionListenerLocal;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.io.UnsupportedEncodingException;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


/**
 * @description: RocketMQ 事务消息
 * @author: MI
 * @email: 448341911@qq.com
 * @createTime: 2025/7/2 18:53
 * @updateUser: MI
 * @updateTime: 2025/7/2 18:53
 * @updateRemark: 修改内容
 * @version: 1.0
 */
public class TransactionProducer {


    public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException, UnsupportedEncodingException {
        // 实例化生产者组名称
        TransactionMQProducer transactionMQProducer = new TransactionMQProducer("tx_producer");
        // 指定name server地址
        transactionMQProducer.setNamesrvAddr("192.168.110.17:9876");
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        transactionMQProducer.setExecutorService(executorService);
        // 本地事务监听
        transactionMQProducer.setTransactionListener(new TransactionListenerLocal());
        // 启动生产者实例
        transactionMQProducer.start();
        for (int i = 0; i < 20; i++) {

            String orderId = UUID.randomUUID().toString();
            String body = "{'operation':'doOrder','orderId':'" + orderId + "','productId':'" + UUID.randomUUID() + "'}";
            Message message = new Message("order_tx_topic",
                    "TagA", orderId, body.getBytes(StandardCharsets.UTF_8));
            transactionMQProducer.sendMessageInTransaction(message, orderId + "&" + i);
            Thread.sleep(1000);

        }

    }
}
